/* Copyright 2012 Dustin Mitchell <dmmitche [at] gmail [dot] com>
 *
 * Licensed under the GPLv3.  See included LICENSE.txt for full details.
 */

#include <stdlib.h>
#include <string.h>

#include "message_queue.h"
#include "iterator.h"
#include "ref_cnt.h"
#include "log.h"



static int message_queue_remove_handler (struct message_queue *queue,
  struct message_handler *handler, int msg_type)
{
  struct list *h_list;
  struct list_iter beg, end;

  if (!hash_exists (&queue->handlers, &msg_type, sizeof (msg_type))) {
    log (LOG_WARNING, "handler not registers with message %d", msg_type);
    pthread_mutex_unlock (&queue->sync);
    return 0;
  }

  h_list = (struct list *)hash_get (&queue->handlers, &msg_type,
    sizeof (msg_type));
  if (!h_list) {
    log (LOG_ERR, "unexpected NULL list in hash");
    pthread_mutex_unlock (&queue->sync);
    return -1;
  }

  if (list_begin (h_list, &beg) == -1 || list_end (h_list, &end) == -1) {
    log (LOG_ERR, "unable to create list iterators");
    pthread_mutex_unlock (&queue->sync);
    return -1;
  }

  for (; !iter_compare (ITER (&beg), ITER (&end)); iter_next (ITER (&beg))) {
    if (iter_deref (ITER (&beg)) == handler
        && list_remove (h_list, &beg) == -1)
      log (LOG_ERR, "unable to remove handler from list");
  }

  return 0;
}



int message_queue_create (struct message_queue *queue)
{
  if (!queue) {
    log (LOG_ERR, "invalid parameter");
    return -1;
  }

  if (pthread_mutex_init (&queue->sync, NULL) != 0) {
    log (LOG_ERR, "unable to create mutex");
    return -1;
  }

  if (hash_create (&queue->handlers, NULL, NULL) == -1) {
    log (LOG_ERR, "unable to create hash");
    pthread_mutex_destroy (&queue->sync);
    return -1;
  }

  return 0;
}

void message_queue_destroy (struct message_queue *queue)
{
  struct hash_iter beg, end;

  if (!queue) {
    log (LOG_ERR, "invalid parameter");
    return;
  }

  if (hash_begin (&queue->handlers, &beg) != -1
      && hash_end (&queue->handlers, &end) != -1) {
    for (; !iter_compare (ITER (&beg), ITER (&end)); iter_next (ITER (&beg))) {
      struct list *l = (struct list *)iter_deref (ITER (&beg));

      if (!l) {
        log (LOG_WARNING, "unexpected NULL value in list");
        continue;
      }

      list_destroy (l);
      free (l);
    }
  }


  hash_destroy (&queue->handlers);

  pthread_mutex_destroy (&queue->sync);
}

int message_queue_register_handler (struct message_queue *queue,
  struct message_handler *handler, int msg_type)
{
  if (!queue || !handler) {
    log (LOG_ERR, "invalid parameter");
    return -1;
  }

  if (pthread_mutex_lock (&queue->sync) != 0) {
    log (LOG_ERR, "unable to create mutex");
    return -1;
  }

  if (!hash_exists (&queue->handlers, &msg_type, sizeof (msg_type))) {
    struct list *type_list = (struct list *)malloc (sizeof (struct list));

    if (!type_list) {
      log (LOG_ERR, "memory allocation failure");
      goto err_out;
    }

    if (list_create (type_list) == -1) {
      log (LOG_ERR, "unable to create list");
      free (type_list);
      goto err_out;
    }

    if (hash_insert (&queue->handlers, &msg_type, sizeof (msg_type), type_list)
        == -1) {
      log (LOG_ERR, "unable to insert value in hash");
      list_destroy (type_list);
      free (type_list);
      goto err_out;
    }
  }

  struct list *type_list = (struct list *)hash_get (&queue->handlers,
    &msg_type, sizeof (msg_type));

  if (!type_list) {
    log (LOG_ERR, "unable to get list from hash");
    goto err_out;
  }

  if (list_push_back (type_list, handler) == -1) {
    log (LOG_ERR, "unable to insert handler in list");
    goto err_out;
  }

  pthread_mutex_unlock (&queue->sync);

  return 0;

err_out:
  pthread_mutex_unlock (&queue->sync);

  return -1;
}

int message_queue_register_handler_multi (struct message_queue *queue,
  struct message_handler *handler, int *msg_types, size_t n_types)
{
  int i;

  if (!msg_types) {
    log (LOG_ERR, "invalid parameter");
    return -1;
  }

  for (i = 0; i < n_types; ++i) {
    if (message_queue_register_handler (queue, handler, msg_types[i]) == -1) {
      log (LOG_ERR, "unable to register handler with queue");
      return -1;
    }
  }

  return 0;
}

int message_queue_unregister_handler (struct message_queue *queue,
  struct message_handler *handler, int msg_type)
{
  if (!queue || !handler) {
    log (LOG_ERR, "invalid parameter");
    return -1;
  }

  if (pthread_mutex_lock (&queue->sync) != 0) {
    log (LOG_ERR, "unable to lock_mutex");
    return -1;
  }

  if (message_queue_remove_handler (queue, handler, msg_type) == -1) {
    log (LOG_ERR, "unable to remove handler");
    pthread_mutex_unlock (&queue->sync);
    return -1;
  }

  pthread_mutex_unlock (&queue->sync);

  return 0;
}

int message_queue_unregister_handler_all (struct message_queue *queue,
  struct message_handler *handler)
{
  struct hash_iter beg, end;

  if (!queue || !handler) {
    log (LOG_ERR, "invalid parameter");
    return -1;
  }

  if (pthread_mutex_lock (&queue->sync) != 0) {
    log (LOG_ERR, "unable to lock_mutex");
    return -1;
  }

  if (hash_begin (&queue->handlers, &beg) == -1
      || hash_end (&queue->handlers, &end) == -1) {
    log (LOG_ERR, "unable to create hash iterators");
    goto out;
  }

  for (; !iter_compare (ITER (&beg), ITER (&end)); iter_next (ITER (&beg))) {
    struct hash_pair *pair = (struct hash_pair *)iter_deref (ITER (&beg));
    int *msg_type;

    if (!pair) {
      log (LOG_ERR, "unexpected NULL in hash");
      goto out;
    }

    msg_type = (int *)pair->key;

    if (message_queue_remove_handler (queue, handler, *msg_type) == -1)
      log (LOG_WARNING, "unable to remove handler for msg %d", *msg_type);
  }

  pthread_mutex_unlock (&queue->sync);

  return 0;

out:
  pthread_mutex_unlock (&queue->sync);

  return -1;
}

int message_queue_post (struct message_queue *queue, const union message *msg)
{
  union message *my_msg = NULL;
  struct ref_cnt *ref;
  struct list *h_list;
  struct list_iter beg, end;

  if (!queue || !msg) {
    log (LOG_ERR, "invalid parameter");
    return -1;
  }

  if (pthread_mutex_lock (&queue->sync) == -1) {
    log (LOG_ERR, "unable to lock mutex");
    return -1;
  }

  if (!hash_exists (&queue->handlers, &msg->header.type,
      sizeof (msg->header.type))) {
    pthread_mutex_unlock (&queue->sync);
    return 0;
  }

  h_list = (struct list *)hash_get (&queue->handlers, &msg->header.type,
    sizeof (msg->header.type));

  if (!h_list) {
    log (LOG_ERR, "unable to get list from hash");
    pthread_mutex_unlock (&queue->sync);
    return -1;
  }

  if (list_empty (h_list)) {
    pthread_mutex_unlock (&queue->sync);
    return 0;
  }

  my_msg = (union message *)malloc (sizeof (union message));

  if (!my_msg) {
    log (LOG_ERR, "memory allocation failure");
    pthread_mutex_unlock (&queue->sync);
    return -1;
  }

  memcpy (my_msg, msg, sizeof (union message));

  ref = (struct ref_cnt *)malloc (sizeof (struct ref_cnt));
  if (!ref) {
    log (LOG_ERR, "memory allocation failure");
    free (my_msg);
    pthread_mutex_unlock (&queue->sync);
    return -1;
  }

  if (ref_cnt_create (ref, my_msg, NULL, free) == -1) {
    log (LOG_ERR, "unable to create ref_cnt object");
    pthread_mutex_unlock (&queue->sync);
    free (ref);
    free (my_msg);
    return -1;
  }

  if (gettimeofday (&my_msg->header.ts, NULL) == -1) {
    log (LOG_ERR, "gettimeofday error");
    goto err_out;
  }

  if (list_begin (h_list, &beg) == -1 || list_end (h_list, &end) == -1) {
    log (LOG_ERR, "unable to create list iterators");
    goto err_out;
  }

  for (; !iter_compare (ITER (&beg), ITER (&end)); iter_next (ITER (&beg))) {
    struct message_handler *handler
      = (struct message_handler *)iter_deref (ITER (&beg));

    if (!handler) {
      log (LOG_WARNING, "unexpected NULL value in list");
      continue;
    }

    if (message_handler_post (handler, ref) == -1) {
      log (LOG_WARNING, "unable to post message to handler");
      continue;
    }
  }

  ref_cnt_deref (ref);
  pthread_mutex_unlock (&queue->sync);

  return 0;

err_out:
  ref_cnt_deref (ref);
  pthread_mutex_unlock (&queue->sync);

  free (ref);

  return -1;
}
